﻿using SuperX.Common.Helper;
using SuperX.Common.Log;
using SuperX.Helper;
using SuperX.SuperSocket;
using SuperX.SuperSocket.Socket.Message;
using SuperX.SuperSocket.Socket.Server;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace SuperX.SuperSocket.Socket.Rpc
{
    public class RpcSever
    {
        private bool isDisposed = false;
        private object _objLock = new object();
        private SocketServer _sockectServer;

        /// <summary>
        /// Key：ClientID，Value：sessionID
        /// </summary>
        private ConcurrentDictionary<string, string> _clientIDandSessionIds { get; set; } = new ConcurrentDictionary<string, string>();

        /// <summary>
        /// 接收命令缓冲区
        /// </summary>
        private readonly ConcurrentDictionary<string, XSocketMessage> ResponseBuffer
            = new ConcurrentDictionary<string, XSocketMessage>();

        private readonly ConcurrentDictionary<string, AutoResetEvent> CommandWaiter
            = new ConcurrentDictionary<string, AutoResetEvent>();

        private ConcurrentDictionary<string, Response<object>> LastResponseLog =
                        new ConcurrentDictionary<string, Response<object>>();

        /// <summary>
        /// 请求的缓存，key:ClientID，value:RpcRequest
        /// </summary>
        private ConcurrentQueue<KeyValuePair<string, RpcRequest>> requestBuff = new ConcurrentQueue<KeyValuePair<string, RpcRequest>>();

        private SocketServerConfig config { get; set; }

        public RpcSever(int port)
            : this(new SocketServerConfig() { ListenPort = port })
        {
        }

        public RpcSever(SocketServerConfig config)
        {
            RegistCommandMethods(GetLocalMethod());//注册当前类的可调用方法
            this.config = config;
            _sockectServer = new SocketServer(config);
            _sockectServer.NewSessionConnected += _socketServer_NewSessionConnected;
            _sockectServer.SessionClosed += _sockectServer_SessionClosed; ;
            _sockectServer.MessageRecevied += SockectServer_MessageRecevied;
            _sockectServer.Error += SockectServer_Error;
            _sockectServer.Start();

            new Thread(SendRpcRequest).Start();
            //Logger.Info($"{nameof(RpcSever)}启动成功");
        }

        private void _sockectServer_SessionClosed(Session session, global::SuperSocket.SocketBase.CloseReason value)
        {
            string clientId = string.Empty;
            foreach (var item in _clientIDandSessionIds)
            {
                if (item.Value == session.SessionID)
                {
                    clientId = item.Key; break;
                }
            }
            _clientIDandSessionIds.TryRemove(clientId, out string sessionID);//移除
        }

        public bool IsRpcClientConnected(string clientID)
        {
            if (string.IsNullOrEmpty(clientID) || !_clientIDandSessionIds.ContainsKey(clientID)) return false;
            var session = _sockectServer.GetSession(_clientIDandSessionIds[clientID]);
            if (session != null && session.Connected)
            {
                return true;
            }
            else
            {
                return false;
            }
        }

        private void _socketServer_NewSessionConnected(Session session)
        {
            string ip = session.RemoteEndPoint.Address.ToString().Trim();
            Logger.Info($"{nameof(RpcSever)}新连接,远程IP：{ip}，本地IP：{session.LocalEndPoint.Address.ToString()}");
        }

        private void SockectServer_Error(object sender, SocketErrorArgs e)
        {
            Logger.Warn($"{nameof(RpcSever)}发生错误：{e.Error},具体内容：{e.Exception.StackTrace}");
        }

        private void SockectServer_MessageRecevied(object sender, MessageArgs e)
        {
            var response = e?.AppMessage?.GetObject<Response<object>>();
            lock (_objLock)
            {
                //1.判断是否是之前的请求返回
                if (response?.RequestId != null)
                {
                    if (CommandWaiter?.ContainsKey(response.RequestId) == true)
                    {
                        ResponseBuffer?.TryAdd(response.RequestId, (XSocketMessage)e?.AppMessage);
                        CommandWaiter.TryRemove(response.RequestId, out AutoResetEvent waiter);
                        waiter.Set();
                    }
                }
                //2.判断是否是要调用server这边的方法，调用应答
                InvokeRequest(e);
            }
        }

        #region Rpc请求

        private void SendRpcRequest()
        {
            while (!isDisposed)
            {
                try
                {
                    while (requestBuff.TryDequeue(out var req))
                    {
                        _sockectServer.GetSession(_clientIDandSessionIds[req.Key])?.Send(JsonHelper.JsonSerialize(req.Value));//发送请求
                        //socket.Send(req);
                        //e.AppSession.Send(JsonHelper.JsonSerialize(new Response<InvokeMessage>()
                        //{
                        //    RequestId = request.RequestId,
                        //    IsSuccess = false,
                        //    Error = new Error(premissionCheckResult.ErrorMsg, 0),
                        //    Data = premissionCheckResult
                        //}));
                    }
                    Thread.Sleep(10);
                }
                catch (Exception)
                {
                }
            }
        }

        /// <summary>
        /// 异步调用
        /// </summary>
        public Task<T> CallAsync<T>(string ip, RpcPara rpcPara, AsyncCallback callback, int timeout = 5000)
        {
            var task = new Task<T>(
                () => Call<T>(ip, rpcPara, timeout)
                );
            task.Start();
            return task;
        }

        /// <summary>
        /// 同步调用返回
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        public T Call<T>(string clientID, RpcPara rpcPara, int timeout = 5000)
        {
            try
            {
                var request = new RpcRequest() { RpcPara = rpcPara };
                var waiter = new AutoResetEvent(false);
                CommandWaiter.TryAdd(request.RequestId, waiter);

                if (rpcPara.Headers == null)
                {
                    rpcPara.Headers = new Dictionary<string, string>();
                }
                rpcPara.Headers.Add("RequestId", request.RequestId);
                //socket.Send(request);
                requestBuff.Enqueue(new KeyValuePair<string, RpcRequest>(clientID, request)); //加到请求命令中
                if (waiter.WaitOne(timeout))
                {
                    if (ResponseBuffer.TryRemove(request.RequestId, out XSocketMessage reponseMsg))
                    {
                        var response = reponseMsg.GetObject<Response<T>>();
                        Response<object> responseObj;
                        if (response != null && response.Data != null)
                        {
                            responseObj = new Response<object>() { Data = response.Data };
                        }
                        else
                        {
                            responseObj = reponseMsg.GetObject<Response<object>>();
                        }
                        LastResponseLog.AddOrUpdate(rpcPara.MethodName, responseObj,
                            (key, oldValue) => responseObj);
                        //if (!LastResponseLog.ContainsKey(rpcPara.MethodName))
                        //{
                        //    LastResponseLog.Add(rpcPara.MethodName, responseObj);
                        //}
                        //else
                        //{
                        //    LastResponseLog[rpcPara.MethodName] = responseObj;
                        //}
                        if (responseObj.IsSuccess != true)
                        {
                            ShowError(response?.Error);
                            return default;
                        }
                        else
                        {
                            if (response == null)
                            {
                                return default;
                            }
                            else
                            {
                                return response.Data;
                                //                    return response.Data;
                            }
                        }
                    }
                }
                else
                {
                    //Reconnect();
                    LastResponseLog[rpcPara.MethodName] = null;
                    lock (_objLock)
                    {
                        CommandWaiter.TryRemove(request.RequestId, out waiter);
                    }

                    var responseObj = new Response<object>()
                    {
                        IsSuccess = false,
                        Error = new Error("rpc request timeout", 10)
                    };
                    Logger.Trace($"rpc request timeout method info:{JsonHelper.JsonSerialize(rpcPara)}");

                    LastResponseLog.AddOrUpdate(rpcPara.MethodName, responseObj,
                        (key, oldValue) => responseObj);
                }
            }
            catch
            {
                //Reconnect();
            }
            return default;
        }

        #endregion Rpc请求

        #region Rpc应答

        private Dictionary<string, MethodInvokeInfo> _commandMethods = new Dictionary<string, MethodInvokeInfo>();

        public void RegistCommandMethods(Dictionary<string, MethodInvokeInfo> methods)
        {
            foreach (var item in methods)
            {
                if (_commandMethods.ContainsKey(item.Key))
                {
                    Logger.Warn($"在RpcServer中，已注册有该方法：{item.Key}");
                    continue;
                }
                _commandMethods.Add(item.Key, item.Value);
            }
        }

        private void InvokeRequest(MessageArgs e)
        {
            Task.Factory.StartNew(() =>
            {
                var request = e.AppMessage.GetObject<RpcRequest>();
                if (request?.RequestId != null)
                {
                    var methodName = request.RpcPara.MethodName;
                    MethodInvokeInfo method = null;
                    if (_commandMethods.ContainsKey(request.RpcPara.MethodName))
                    {
                        method = _commandMethods[request.RpcPara.MethodName];
                    }
                    Response<object> response = null;

                    if (method != null)
                    {
                        var result = method.Invoke(request.RpcPara.Paras);//调用函数

                        if (result is Response<object> tempresponse)
                        {
                            response = new Response<object>(tempresponse) { RequestId = request.RequestId };
                        }
                        else
                        {
                            response = new Response<object>() { RequestId = request.RequestId, Data = result };
                        }
                    }
                    else
                    {
                        response = new Response<object>()
                        {
                            IsSuccess = false,
                            RequestId = request.RequestId,
                            Error = new Error($"Gateway Method not found:{methodName}", 0)
                        };
                    }
                    e.AppSession.Send(JsonHelper.JsonSerialize(response));
                }
            });
        }

        #endregion Rpc应答

        public void Dispose()
        {
            isDisposed = true;
            _sockectServer?.Stop();
            _sockectServer.NewSessionConnected -= _socketServer_NewSessionConnected;
            _sockectServer.MessageRecevied -= SockectServer_MessageRecevied;
            _sockectServer.Error -= SockectServer_Error;
        }

        private void ShowError(Error error)
        {
            if (error != null)
            {
                if (error.InnerError != null)
                {
                    ShowError(error.InnerError);
                }
                else
                {
                    Logger.Warn(error.Message);
                }
            }
        }

        public Response<object> GetLastRequestResult(string methodName)
        {
            if (LastResponseLog.ContainsKey(methodName))
            {
                return LastResponseLog[methodName];
            }
            return null;
        }

        /// <summary>
        /// 注册当前类的可调用方法
        /// </summary>
        /// <returns></returns>
        private Dictionary<string, MethodInvokeInfo> GetLocalMethod()
        {
            var methods = new List<MethodInvokeInfo>();
            methods.AddRange(GetType().GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.DeclaredOnly)
                           .Select(t => new MethodInvokeInfo()
                           {
                               ExecObj = this,
                               MethodName = t.Name,
                               MethodInfo = t,
                               Handler = DynamicEmitCalls.GetMethodInvoker(t)
                           }));
            return methods.ToDictionary(key => key.MethodName, value => value);
        }

        /// <summary>
        ///
        /// </summary>
        /// <param name="clientId">客户端ID</param>
        /// <param name="sessionId">会话ID</param>
        /// <returns></returns>
        public bool RegisterRpcClient(string clientId, string sessionId)
        {
            if (string.IsNullOrEmpty(clientId) || string.IsNullOrEmpty(sessionId)) return false;
            _clientIDandSessionIds.TryAdd(clientId, sessionId);
            return true;
        }
    }

    public class MethodInvokeInfo
    {
        /// <summary>
        /// 方法名
        /// </summary>
        public string MethodName { get; set; }

        /// <summary>
        /// 执行对象
        /// </summary>
        public object ExecObj { get; set; }

        /// <summary>
        /// 方法信息
        /// </summary>
        public MethodInfo MethodInfo { get; set; }

        /// <summary>
        ///
        /// </summary>
        public FastInvokeHandler Handler { get; set; }

        public object Invoke(object[] args)
        {
            return Handler.Invoke(ExecObj, args);
        }
    }
}